package com.itcast.zk.locker.service;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.CountDownLatch;

/**
 * <p>Description: 排他锁</p>
 *
 * @author
 * @version 1.0
 * <p>Copyright:Copyright(c)2020</p>
 * @date
 */
@Service("exclusiveLocker")
public class ExclusiveLocker implements InitializingBean, ILocker {
    private static final Logger LOGGER = LoggerFactory.getLogger(ExclusiveLocker.class);
    
    private static final String ROOT_LOCK_PATH = "exclusive_lock";
    
    private CountDownLatch countDownLatch = new CountDownLatch(1);
    
    @Autowired
    private CuratorFramework curatorFramework;
    
    /**
     * 排它锁的申请
     *
     * @param path
     * @param type
     * @return
     */
    @Override
    public String acquireDistributedLock(String path, String type) {
        String keyPath = "/" + ROOT_LOCK_PATH + "/" + path;
        while (true) {
            try {
                curatorFramework.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.EPHEMERAL)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(keyPath);
                LOGGER.info("[ExclusiveLockByCurator acquireDistributedLock] success to acquire lock for path: {}", keyPath);
                return "true";
            } catch (Exception e) {
                LOGGER.info("[ExclusiveLockByCurator acquireDistributedLock] failed to acquire lock for path: {}", keyPath);
                LOGGER.info("try again... ...");
                try {
                    if (countDownLatch.getCount() <= 0) {
                        countDownLatch = new CountDownLatch(1);
                    }
                    countDownLatch.await();
                } catch (Exception e1) {
                    LOGGER.error("[ExclusiveLockByCurator acquireDistributedLock] error", e1);
                }
            }
        }
        
    }
    
    @Override
    public String acquireDistributedLockNoWait(String path, String type) {
        String keyPath = "/" + ROOT_LOCK_PATH + "/" + path;
        try {
            curatorFramework.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                    .forPath(keyPath);
            LOGGER.info("[ExclusiveLockByCurator acquireDistributedLock] success to acquire lock for path: {}", keyPath);
            return "true";
        } catch (Exception e) {
            LOGGER.info("[ExclusiveLockByCurator acquireDistributedLock] failed to acquire lock for path: {}", keyPath);
        }
        
        return null;
    }
    
    @Override
    public boolean releaseDistributedLock(String path) {
        try {
            String keyPath = "/" + ROOT_LOCK_PATH + "/" + path;
            if (curatorFramework.checkExists().forPath(keyPath) != null) {
                curatorFramework.delete().forPath(keyPath);
                return true;
            }
        } catch (Exception e) {
            LOGGER.error("[ExclusiveLockByCurator acquireDistributedLock] failed to release lock");
        }
        return false;
    }
    
    private void addWatcher(String path) throws Exception {
        String keyPath;
        if (path.equals(ROOT_LOCK_PATH)) {
            keyPath = "/" + path;
        } else {
            keyPath = "/" + ROOT_LOCK_PATH + "/" + path;
        }
        final PathChildrenCache cache = new PathChildrenCache(curatorFramework, keyPath, false);
        cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        cache.getListenable().addListener((curatorFramework, event) -> {
            if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                String oldPath = event.getData().getPath();
                LOGGER.info("[ExclusiveLockByCurator acquireDistributedLock] success to release lock for path: {}", oldPath);
                if (oldPath.contains(path)) {
                    // 子节点被删除 通知其他客户端
                    countDownLatch.countDown();
                }
            }
        });
    }
    
    @Override
    public void afterPropertiesSet() throws Exception {
        curatorFramework = curatorFramework.usingNamespace("lock-namespace");
        String path = "/" + ROOT_LOCK_PATH;
        try {
            if (curatorFramework.checkExists().forPath(path) == null) {
                curatorFramework.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                        .forPath(path);
            }
            addWatcher(ROOT_LOCK_PATH);
            LOGGER.info("root path 的 watcher 事件创建成功");
        } catch (Exception e) {
            LOGGER.error("connect zookeeper fail，please check the log >> {}", e.getMessage(), e);
        }
    }
    
}
